package pre

import com.alibaba.fastjson.{JSON, JSONException, JSONObject}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import util.SparkUtil
import java.util

import bean.AppLogBean
import ch.hsr.geohash.GeoHash
import org.apache.spark.broadcast.Broadcast

/**
 * APP埋点日志预处理
 */
object AppLogDataPreProcess {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkUtil.getSparkSession(this.getClass.getSimpleName)
    import spark.implicits._

    // 加载geo地理位置字典，然后收集到driver端并广播到executor端
    val geoDF: DataFrame = spark.read.parquet("data/dict/geo_dict/output")
    val geoMap: collection.Map[String, (String, String, String)] = geoDF.rdd.map({
      case Row(geoCode: String, province: String, city: String, district: String) => (geoCode, (province, city, district))
    }).collectAsMap()
    val bc_geo: Broadcast[collection.Map[String, (String, String, String)]] = spark.sparkContext.broadcast(geoMap)

    // 加载idmp映射字典，然后收集到driver端并广播到executor端
    val idmpDF: DataFrame = spark.read.parquet("data/idmp/2020-01-12")
    val idmpMap: collection.Map[Long, Long] = idmpDF.rdd.map({
      case Row(biaoshi_hashcode: Long, guid: Long) => (biaoshi_hashcode, guid)
    }).collectAsMap()
    val bc_idmp: Broadcast[collection.Map[Long, Long]] = spark.sparkContext.broadcast(idmpMap)

    // 加载当日的APP埋点日志数据
    val appLog: Dataset[String] = spark.read.textFile("E:\\doit项目资料\\DW12-文档资料\\day03\\yiee_logs\\2020-01-12")
    val beanDS: Dataset[AppLogBean] = appLog.map(line => {
      extractAll(line)
    }).filter(_ != null) // 过滤掉不符合要求的数据
    val resBeans: DataFrame = beanDS.mapPartitions(iter => {
      val geoDict: collection.Map[String, (String, String, String)] = bc_geo.value
      val idmpDict: collection.Map[Long, Long] = bc_idmp.value
      iter.map(bean => {
        val lng: Double = bean.longtitude
        val lat: Double = bean.latitude
        val geoCode: String = GeoHash.geoHashStringWithCharacterPrecision(lat, lng, 5)
        val maybeTuple: Option[(String, String, String)] = geoDict.get(geoCode)
        if (maybeTuple.isDefined) {
          bean.province = maybeTuple.get._1
          bean.city = maybeTuple.get._2
          bean.district = maybeTuple.get._3
        }
        val ids: Array[String] = Array(bean.uid, bean.imei, bean.imsi, bean.mac, bean.androidId, bean.uuid)
        var find: Boolean = false
        for (elem <- ids if !find) {
          val maybeLong: Option[Long] = idmpDict.get(elem.hashCode.toLong)
          if (maybeLong.isDefined) {
            bean.guid = maybeLong.get
            find = true
          }
        }
        bean
      })
    }).filter(bean => bean.guid != 0).toDF()

    resBeans.write.parquet("data/appLog_processed/2020-01-12")

  }

  def extractAll(line: String): AppLogBean = {
    try {
      // 将一行数据解析成json对象
      val jsonObj: JSONObject = JSON.parseObject(line) // 总对象
      val eventid: String = jsonObj.getString("eventid")

      val eventObj: JSONObject = jsonObj.getJSONObject("event") // event对象
      import scala.collection.JavaConversions._
      val eventMap: Map[String, String] = eventObj.getInnerMap.asInstanceOf[util.Map[String, String]].toMap

      val userObj: JSONObject = jsonObj.getJSONObject("user") // user对象
      val uid = userObj.getString("uid")
      val phoneObj: JSONObject = userObj.getJSONObject("phone") // phone对象
      val imei: String = phoneObj.getString("imei")
      val mac: String = phoneObj.getString("mac")
      val imsi: String = phoneObj.getString("imsi")
      val osName: String = phoneObj.getString("osName")
      val osVer: String = phoneObj.getString("osVer")
      val androidId: String = phoneObj.getString("androidId")
      val resolution: String = phoneObj.getString("resolution")
      val deviceType: String = phoneObj.getString("deviceType")
      val deviceId: String = phoneObj.getString("deviceId")
      val uuid: String = phoneObj.getString("uuid")

      val appObj: JSONObject = userObj.getJSONObject("app") // app对象
      val appid: String = appObj.getString("appid")
      val appVer: String = appObj.getString("appVer")
      val release_ch: String = appObj.getString("release_ch")
      val promotion_ch: String = appObj.getString("promotion_ch")

      val locObj: JSONObject = userObj.getJSONObject("loc") // loc对象
      val longtitude: Double = locObj.getDouble("longtitude")
      val latitude: Double = locObj.getDouble("latitude")
      val carrier: String = locObj.getString("carrier")
      val netType: String = locObj.getString("netType")
      val cid_sn: String = locObj.getString("cid_sn")
      val ip: String = locObj.getString("ip")

      val sessionId: String = userObj.getString("sessionId")
      val timestamp: Long = jsonObj.getString("timestamp").toLong
      // 返回结果，如果符合要求就返回一个AppLogBean，否则返回一个null
      if (uid.equals("") && imei.equals("") && uuid.equals("") && mac.equals("") && androidId.equals("") && imsi.equals("")) {
        null
      } else if (eventObj == null || eventid.equals("") || sessionId.equals("")) {
        null
      } else {
        AppLogBean(0, eventid, eventMap, uid, imei, mac, imsi, osName, osVer, androidId, resolution, deviceType, deviceId, uuid,
          appid, appVer, release_ch, promotion_ch, longtitude, latitude, carrier, netType, cid_sn, ip, sessionId, timestamp)
      }
    } catch {
      case _: JSONException => null
    }
  }
}
